package com.leyantech.ecp.agenx.translator.consumer;

import com.leyantech.ai.DialogueReply;
import com.leyantech.chaos.kafka.KafkaConsumerProcessor;
import com.leyantech.ecp.agenx.common.converter.output.DialogueReplyConverter;
import com.leyantech.ecp.agenx.common.sender.Dispatcher;
import com.leyantech.ecp.agenx.common.utils.PbPrintHelper;
import com.leyantech.ecp.agenx.translator.filter.dialog.DialogueReplyFilter;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-01.
 */
@Slf4j
@Singleton
public class DialogueMessageConsumer implements KafkaConsumerProcessor<String, DialogueReply> {

  private final Dispatcher dispatcher;

  private final DialogueReplyConverter dialogueReplyConverter;

  private final DialogueReplyFilter filter;

  @Inject
  public DialogueMessageConsumer(Dispatcher dispatcher,
      DialogueReplyConverter dialogueReplyConverter,
      DialogueReplyFilter filter) {
    this.dispatcher = dispatcher;
    this.dialogueReplyConverter = dialogueReplyConverter;
    this.filter = filter;
  }

  @Override
  public void accept(ConsumerRecord<String, DialogueReply> consumerRecord) {
    final DialogueReply dialogueReply = consumerRecord.value();

    if (!filter.test(dialogueReply)) {
      return;
    }

    log.info("received dialogueReply {}", PbPrintHelper.lazyToJson(dialogueReply));

    final JsonObject reply = dialogueReplyConverter.apply(dialogueReply);

    final String storeId = dialogueReply.getStoreId();
    final String assistantId = dialogueReply.getAssistantId();

    dispatcher.sendToConnector(storeId, assistantId, reply.toString());
  }

}
